package gossip;

import lombok.Getter;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Getter
public class GossipCore {

    private final MembershipTable membershipTable;
    private final GossipTransport transport;
    private final FailureDetector failureDetector;

    public GossipCore(int port) {
        this.membershipTable = new MembershipTable(port);
        this.transport = null; // 将在外部设置
        this.failureDetector = new FailureDetector(this);
    }

    public void start(ScheduledExecutorService scheduler) {
        // 每500ms执行一次gossip传播
        scheduler.scheduleAtFixedRate(this::doGossip, 0, 500, TimeUnit.MILLISECONDS);

        // 每1秒执行一次故障检测
        scheduler.scheduleAtFixedRate(failureDetector::check, 0, 1, TimeUnit.SECONDS);
    }

    private void doGossip() {
        List<MemberState> targets = membershipTable.selectRandomTargets(3);
        for (MemberState target : targets) {
            sendGossipDigest(target.getAddress());
        }
    }

    private void sendGossipDigest(InetSocketAddress target) {
        Map<String, Long> digest = new HashMap<>();
        membershipTable.getMembers().forEach((id, state) -> digest.put(id, state.getHeartbeat()));

        String message = "DIGEST:" + serializeDigest(digest);
        transport.send(target, message);
    }

    private String serializeDigest(Map<String, Long> digest) {
        return digest.entrySet().stream()
                .map(e -> e.getKey() + "=" + e.getValue())
                .collect(Collectors.joining(","));
    }

    public void handleMessage(SocketAddress sender, String message) {
        if (message.startsWith("DIGEST:")) {
            handleGossipDigest(sender, message.substring(7));
        } else if (message.startsWith("SYNC:")) {
            handleFullSync(message.substring(5));
        }
    }

    private void handleGossipDigest(SocketAddress sender, String digestStr) {
        System.out.printf("[%s] Received DIGEST from %s: %s%n", "", sender, digestStr);
        // 解析摘要
        Map<String, Long> remoteDigest = parseDigest(digestStr);

        // 生成本地摘要
        Map<String, Long> localDigest = new HashMap<>();
        membershipTable.getMembers().forEach((id, state) ->
                localDigest.put(id, state.getHeartbeat()));

        // 找出需要同步的节点
        List<String> needSync = new ArrayList<>();
        for (Map.Entry<String, Long> entry : remoteDigest.entrySet()) {
            Long localHeartbeat = localDigest.get(entry.getKey());
            if (localHeartbeat == null || localHeartbeat < entry.getValue()) {
                needSync.add(entry.getKey());
            }
        }

        // 发送需要同步的节点请求
        String response = "SYNC:" + String.join(",", needSync);
        assert transport != null;
        transport.send((InetSocketAddress) sender, response);
    }

    private void handleFullSync(String nodeList) {
        if (nodeList.isEmpty()) return;

        // 获取请求的节点状态
        Map<String, MemberState> states = new HashMap<>();
        for (String nodeId : nodeList.split(",")) {
            MemberState state = membershipTable.getMembers().get(nodeId);
            if (state != null) {
                states.put(nodeId, state);
            }
        }

        // 发送完整状态
        String syncData = "FULLSYNC:" + serializeStates(states);
        // 在实际实现中，这里应该发送给请求方，但需要会话保持
    }

    private String serializeStates(Map<String, MemberState> states) {
        // 简化的序列化实现
        return states.entrySet().stream()
                .map(e -> e.getKey() + ":" + e.getValue().getHeartbeat())
                .collect(Collectors.joining(","));
    }

    private Map<String, Long> parseDigest(String digestStr) {
        Map<String, Long> digest = new HashMap<>();
        for (String part : digestStr.split(",")) {
            String[] kv = part.split("=");
            if (kv.length == 2) {
                digest.put(kv[0], Long.parseLong(kv[1]));
            }
        }
        return digest;
    }

    public void join(InetSocketAddress seedNode) {
        sendGossipDigest(seedNode);
    }

    public void updateLocalMetadata(String key, String value) {
        membershipTable.updateLocalMetadata(key, value);
    }

    public MembershipTable getMembershipTable() {
        return membershipTable;
    }
}
